package com.bw.yk10.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

public class MyKafkaUtil {

    public FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId) {
        Properties properties = new Properties();

        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.18.105:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        FlinkKafkaConsumer<String> flinkKafkaConsumer=new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), properties);
        flinkKafkaConsumer.setStartFromEarliest();
        return flinkKafkaConsumer;
    }

    public FlinkKafkaProducer<String> getKafkaProducer(String topic) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.18.105:9092");
        return new FlinkKafkaProducer<>(topic, new SimpleStringSchema(), properties);
    }

}
